package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackInfoMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.login.GuiGuLogin;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackInfo;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.client.RedisClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumAttributeValueMapper albumAttributeValueMapper;

	@Autowired
	private AlbumStatMapper albumStatMapper;

	@Autowired
	private TrackInfoMapper trackInfoMapper;

	@Autowired
	private AlbumAttributeValueService albumAttributeValueService;

	@Autowired
	private KafkaService kafkaService;

	@Autowired
	private RedisTemplate redisTemplate;

	@Autowired
	private RedissonClient redissonClient;

	/**
	 * 保存专辑
	 *
	 * @param userId
	 * @param albumInfoVo
	 */
	@Override
	@Transactional(rollbackFor = Exception.class) // 默认发送RunTimeException或者Error错误回滚
	public void saveAlbumInfo(Long userId,AlbumInfoVo albumInfoVo) {
		// 1.将提交的专辑信息封装到albumInfoVo对象，向专辑表中添加一条记录
		 // 1.1 将提交专辑的Vo转换为Po对象
		AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo,AlbumInfo.class);
		// 1.2 为专辑中的属性赋值 用户ID，包含声音数量、免费试听集数、状态等
		albumInfo.setUserId(userId);
		albumInfo.setIncludeTrackCount(0);
		albumInfo.setTracksForFree(5);
		albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
		 // 1.3 保存专辑 得到专辑ID
		albumInfoMapper.insert(albumInfo);
		 // 得到该条专辑的Id
		Long albumInfoId = albumInfo.getId();
		// 2.将提交专辑标签封装到专辑标签集合对象，向专辑标签表添加若干条记录
		List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
		if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)){
			albumAttributeValueVoList.stream()
					.forEach(albumAttributeValueVo ->{
						// 2.1 获取专辑标签VO 转为专辑标签PO集合
						AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
						// 2.2 专辑标签关联专辑ID，保存
						albumAttributeValue.setAlbumId(albumInfoId);
						// 2.3 保存专辑标签
						albumAttributeValueMapper.insert(albumAttributeValue);
					});
		}

		// 3.固定为专辑保存四条统计信息
		this.saveAlbumStat(albumInfoId,SystemConstant.ALBUM_STAT_PLAY,0);
		this.saveAlbumStat(albumInfoId,SystemConstant.ALBUM_STAT_SUBSCRIBE,0);
		this.saveAlbumStat(albumInfoId,SystemConstant.ALBUM_STAT_BUY,0);
		this.saveAlbumStat(albumInfoId,SystemConstant.ALBUM_STAT_COMMENT,0);

		//4.TODO 调用内容审核接口（第三方阿里云）对专辑内容（封面、文字）

		//5.审核通过后发送上架专辑消息到Kafka
		if (true && "1".equals(albumInfo.getIsOpen())){
			kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_UPPER,albumInfoId.toString());
		}
	}

	/**
	 * 保存专辑统计信息
	 *
	 * @param albumId  专辑ID
	 * @param statType 统计类型 0401-播放量 0402-订阅量 0403-购买量 0403-评论数
	 * @param num      数值
	 */
	@Override
	public void saveAlbumStat(Long albumId, String statType, Integer num) {
		AlbumStat albumStat = new AlbumStat();
		albumStat.setAlbumId(albumId);
		albumStat.setStatType(statType);
		albumStat.setAlbumId(albumId);
		albumStatMapper.insert(albumStat);
	}

	/**
	 * 该接口必须要求用户登录后才能调用，获取用户ID关联到专辑 暂时将当前用户ID固定为：1
	 * 分页查询当前用户专辑列表
	 *
	 * @param pageInfo           分页对象
	 * @param albumInfoQuery 查询条件
	 * @return 分页对象
	 */
	@Override

	public Page<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> pageInfo, AlbumInfoQuery albumInfoQuery) {
		// 获取专辑持久层对象，调用持久层对象方法（动态sql + 分页）
		return albumInfoMapper.getUserAlbumPage(pageInfo,albumInfoQuery);
	}

	/**
	 * 根据ID删除专辑
	 * @param id
	 */
	@Override
	public void removeAlbumInfo(Long id) {
		// 1.判断该专辑是否包含声音，如果包含则拒绝删除
		LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
		queryWrapper.eq(TrackInfo::getAlbumId,id);
		Long selectCount = trackInfoMapper.selectCount(queryWrapper);
		if (selectCount > 0){
			throw new GuiguException(400,"该专辑下仍然关联" + selectCount + "个声音");
		}
		// 2.根据主键删除专辑表（逻辑删除）
		albumInfoMapper.deleteById(id);
		// 3.根据专辑ID删除专辑标签记录（逻辑删除）
		LambdaQueryWrapper<AlbumAttributeValue> attributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
		attributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId,id);
		albumAttributeValueMapper.delete(attributeValueLambdaQueryWrapper);
		// 4.根据专辑ID删除专辑统计记录（逻辑删除）
		LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
		albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId,id);
		albumStatMapper.delete(albumStatLambdaQueryWrapper);

		//TODO 同步删除索引库中专辑文档
		kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
	}

	/**
	 * 根据ID查询专辑信息（包含专辑标签列表）
	 * [自定义分布式锁避免缓存击穿]
	 * @param id
	 * @return
	 */
	@Override
	public AlbumInfo getAlbumInfo(Long id) {
		// 1. 优先从缓存Redis中获取业务数据
		// 1.1 构建专辑信息业务数据Key
		String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
		// 1.2 从Redis中获取业务数据
		AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
		// 1.3 命中缓存则执行返回，未命中缓存执行第二步
		if (albumInfo != null){
			return albumInfo;
		}

		// 2. 尝试获取分布式锁
		// 2.1 构建锁的Key
		String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
		// 2.2 构建锁的Value 采用唯一ID避免不同线程互相释放锁
		String lockVal = IdUtil.fastSimpleUUID();
		// 2.3 采用set ex nx命令尝试获取锁
		Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockVal, 5, TimeUnit.SECONDS);
		if (flag){
			try {
				// 3. 获取锁成功执行业务代码
				// 3.1 从数据库中查询业务数据
				albumInfo = this.getAlbumInfoFromDB(id);
				// 3.2 将查询业务数据放入缓存
				int randomTTL = RandomUtil.randomInt(200,2000);
				redisTemplate.opsForValue().set(dataKey,albumInfo,RedisConstant.ALBUM_TIMEOUT + randomTTL,TimeUnit.SECONDS);
				// 3.3 将结果返回
				return albumInfo;
			} finally {
				// 4. 释放锁
				String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
						"then\n" +
						"    return redis.call(\"del\",KEYS[1])\n" +
						"else\n" +
						"    return 0\n" +
						"end";
				DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
				redisScript.setScriptText(scriptText);
				redisScript.setResultType(Long.class);
				redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockVal);
			}
		}else {
			// 5. 获取锁失败执行业务自旋
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				log.error("获取锁失败，自旋：{}，线程名称：{}", Thread.currentThread().getId(), Thread.currentThread().getName());
				throw new RuntimeException(e);
			}
			return this.getAlbumInfo(id);
		}

	}

	/**
	 * 根据ID查询专辑信息（包含专辑标签列表）
	 * [Redission分布式锁避免缓存击穿]
	 * @param id
	 * @return
	 */
	@Override
	public AlbumInfo getAlbumInfoForRedisson(Long id) {
		// 1. 优先从缓存获取业务数据
		// 1.1 构建专辑信息业务Key
		String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
		// 1.2 从Redis中获取业务数据
		AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
		// 1.3 命中缓存则返回 未命中则执行第二步
		if (albumInfo != null){
			return albumInfo;
		}

		// 2. 获取分布式锁，看门狗实现锁自动续期
		// 2.1 构建锁的Key
		String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
		// 2.2 创建锁对象
		RLock lock = redissonClient.getLock(lockKey);
		// 2.3 获取锁 线程阻塞到获取锁成功为止
		lock.lock();

		try {
			// 3. 执行业务代码
			// 3.1 从DB获取数据逻辑
			albumInfo = this.getAlbumInfoFromDB(id);
			// 3.2 将查询结果换入缓存并还回
			int randomTTl = RandomUtil.randomInt(200,2000);
			redisTemplate.opsForValue().set(dataKey,albumInfo,RedisConstant.ALBUM_TIMEOUT + randomTTl,TimeUnit.SECONDS);
			return albumInfo;
		} finally {
			// 3.3 业务执行完毕释放锁
			lock.unlock();
		}
	}

	/**
	 * 根据ID查询专辑信息（包含专辑标签列表）
	 * [从数据库查询]
	 * @param id
	 * @return
	 */
	@Override
	public AlbumInfo getAlbumInfoFromDB(Long id) {
		// 1.根据专辑主键ID查询专辑信息
		AlbumInfo albumInfo = albumInfoMapper.selectById(id);
		Assert.notNull(albumInfo,"专辑信息：{}不存在",id);
		// 2.根据专辑主键ID查询专辑标签列表，将标签列表封装到专辑对象中
		LambdaQueryWrapper<AlbumAttributeValue> attributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
		attributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId,id);
		List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(attributeValueLambdaQueryWrapper);
		if (CollectionUtil.isNotEmpty(albumAttributeValueList)){
			albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
		}
		return albumInfo;
	}

	/**
	 * 修改专辑
	 * @param id
	 * @param albumInfoVo
	 */
	@Override
	public void updateAlbumInfo(Long id, AlbumInfoVo albumInfoVo) {
		// 1.修改专辑信息
		 // 1.1 将提交专辑信息VO转为PO对象（手动封装专辑id）
		AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
		albumInfo.setId(id);
		// 1.2 调用持久层修改
		albumInfoMapper.updateById(albumInfo);

		// 2.调用持久层修改
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)){
			// 2.1 根据专辑id先删除专辑标签关系表
			LambdaQueryWrapper<AlbumAttributeValue> attributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
			attributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId,id);
			albumAttributeValueMapper.delete(attributeValueLambdaQueryWrapper);
			// 2.2 新增专辑关系表记录
			List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
				AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
				albumAttributeValue.setAlbumId(id);
				return albumAttributeValue;
			}).collect(Collectors.toList());
			// 2.3 批量新增专辑标签关系
			albumAttributeValueService.saveBatch(albumAttributeValueList);

			//TODO:如果是开放专辑自动将专辑同步到ES索引库中
			if ("1".equals(albumInfo.getIsOpen())){
				kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_UPPER,id.toString());
			}else {
				kafkaService.sendKafkaMessage(KafkaConstant.QUEUE_ALBUM_UPPER,id.toString());
			}

		}
	}

	/**
	 * 获取当前用户全部专辑列表
	 * 根据userId，查询专辑Id 与专辑标题，然后按照专辑Id 进行降序排列
	 * @param userId
	 * @return
	 */
	@Override
	public List<AlbumInfo> getUserAllAlbumList(Long userId) {
		// 1.构建查询条件对象
		LambdaQueryWrapper<AlbumInfo> albumInfoLambdaQueryWrapper = new LambdaQueryWrapper<>();
		 // 1.1 设置查询条件：用户ID
		albumInfoLambdaQueryWrapper.eq(AlbumInfo::getUserId,userId);
		 // 1.2 设置排序字段
		albumInfoLambdaQueryWrapper.orderByDesc(AlbumInfo::getId);
		 // 1.3 设置查询字段 设置select部分
		albumInfoLambdaQueryWrapper.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle);
		 // 1.4 设置限制返回记录数 设置Limt 限制展示200条
		albumInfoLambdaQueryWrapper.last("limit 200");

		// 2.执行查询
		List<AlbumInfo> list = albumInfoMapper.selectList(albumInfoLambdaQueryWrapper);
		return list;
	}

	/**
	 * 根据专辑ID获取专辑统计信息
	 * @param albumId
	 * @return
	 */
	@Override
	public AlbumStatVo getAlbumStatVo(Long albumId) {
		return albumInfoMapper.selectAlbumStatVo(albumId);
	}

}
